package cn.doitedu.kafka.rebalance;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class RebalanceConsumer2 {
    public static void main(String[] args) {
        Properties properties=new Properties();
        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer",StringDeserializer.class.getName());
        properties.put("bootstrap.servers","doit01:9092,doit02:9092,doit03:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"gp-1");
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");


        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Arrays.asList("tpc-b","tpc-c"));
        while(true){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
            poll.forEach(r-> System.out.println(r.key() + " --> " + r.value()));
        }
    }

}
